package com.shujia.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo01WcntOnStreaming {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo01WcntOnStreaming")
    conf.setMaster("local[2]") // 一个线程是无法进行实时计算的，因为接收数据需要一直占用一个线程

    val sc: SparkContext = new SparkContext(conf)

    // 构建StreamingContext
    // 需要接收两个参数：SparkContext、批次大小（时间，每个多久会产生一个job）
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))

    // 在Spark Streaming的环境中 主要操作的数据结构是DStream
    // 通过nc -lk 8888命令在端口上启动一个Socket Server
    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    lineDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .reduceByKey(_ + _) // 默认每个批次的结果互不影响
      .print()

    // 启动实时程序
    ssc.start()
    // 等待程序停止
    ssc.awaitTermination()
    // 关闭程序
    ssc.stop()


  }

}
